package xyz.xuw.storm.workcount.topology;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import xyz.xuw.storm.workcount.spout.WordSpout;
import xyz.xuw.storm.workcount.bolt.WordBolt;
import xyz.xuw.storm.workcount.bolt.WordCounterBolt;
import xyz.xuw.storm.workcount.model.MyMessage;

/**
 * Created by xuw on 2016/11/10.
 */
public class WordCountTopology {
    public static void main(String[] args) throws InterruptedException {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("word-reader", new WordSpout());
        builder.setBolt("word-bolt", new WordBolt()).shuffleGrouping("word-reader");
        builder.setBolt("word-counter", new WordCounterBolt(), 2)
                .fieldsGrouping("word-bolt", new Fields("word"));
        Config config = new Config();
        config.put("wordsFile", WordCountTopology.class.getResource("/word.txt").getPath().toString());
        config.registerSerialization(MyMessage.class);
        config.setDebug(false);
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("getting-Started-Topologie", config, builder.createTopology());
        Thread.sleep(10000);
        cluster.shutdown();
    }
}
